package com.zhichat.web.socket.netty.handler;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.util.StrUtil;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.google.common.collect.Lists;
import com.zhichat.dao.ApplyInfoDao;
import com.zhichat.dao.ChatUserSessionDao;
import com.zhichat.dao.ContactInfoDao;
import com.zhichat.dao.UserInfoDao;
import com.zhichat.document.ChatMsgDoc;
import com.zhichat.entity.ApplyInfoEntity;
import com.zhichat.entity.ContactInfoEntity;
import com.zhichat.entity.UserInfoEntity;
import com.zhichat.infrastructure.Enums.*;
import com.zhichat.infrastructure.config.ZhiChatGlobalUtil;
import com.zhichat.infrastructure.constants.Constants;
import com.zhichat.infrastructure.exception.http.CommonException;
import com.zhichat.infrastructure.redis.Keys;
import com.zhichat.infrastructure.redis.RedisRepository;
import com.zhichat.infrastructure.redis.rediskey.Key;
import com.zhichat.infrastructure.redis.rediskey.ZhiChat;
import com.zhichat.infrastructure.util.ThreadLocalCache;
import com.zhichat.web.dto.UserToken;
import com.zhichat.web.socket.ChannelContextCommon;
import com.zhichat.web.socket.GlobalMsgHandle;
import com.zhichat.web.vo.ws.MsgVO;
import com.zhichat.web.vo.ws.UserChatMsgVO;
import com.zhichat.web.vo.ws.UserInitVO;
import com.zhichat.web.vo.ws.UserSessionInfoVO;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
import io.netty.handler.timeout.IdleStateEvent;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;

import java.net.URLDecoder;
import java.time.Duration;
import java.util.List;
import java.util.Objects;
import java.util.stream.Collectors;

/**
 * @ClassName: WsTextMsgHandler
 * @Description: 文本消息处理handler（只处理文本消息，如果有视频、图片等非文本消息，可以走子协议去处理）
 * @Author: 团长九哥 https://space.bilibili.com/602197686
 * @CreateDate: 2024/11/29 18:09
 */
@Slf4j
@Component
@AllArgsConstructor
@ChannelHandler.Sharable
public class WsTextMsgHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {

    private final RedisRepository redisRepository;
    private final ChannelContextCommon channelContextCommon;
    private final UserInfoDao userInfoDao;
    private final ChatUserSessionDao chatUserSessionDao;
    private final MongoTemplate mongoTemplate;
    private final ApplyInfoDao applyInfoDao;
    private final GlobalMsgHandle globalMsgHandle;
    private final ZhiChatGlobalUtil zhiChatGlobalUtil;
    private final ContactInfoDao contactInfoDao;

    /**
     * 有新消息时，触发
     *
     * @param ctx the {@link ChannelHandlerContext} which this {@link SimpleChannelInboundHandler}
     *            belongs to
     * @param msg the message to handle
     * @throws Exception
     */
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception {
        Long curTime = System.currentTimeMillis();
        Channel channel = ctx.channel();
        String channelId = channel.id().toString();
        Long userId = ThreadLocalCache.getUserId();
        log.debug("用户:{}，channelId:{}，收到消息:{}", userId, channelId, msg.text());
        // 设置心跳 -> redis（在登录的时候，会校验心跳，如果心跳存在，则不允许登录。主要针对已在设备登录，不允许其他地方登录）
        Key heartbeatKey = new ZhiChat(Keys.USER_HEARTBEAT, Duration.ofSeconds(Constants.READER_IDLE_TIME));
        redisRepository.set(heartbeatKey, curTime, userId);
    }

    /**
     * 有新连接时，触发
     *
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        log.info("有新连接！ channelId:{}", ctx.channel().id().toString());
        super.channelActive(ctx);
    }

    /**
     * 连接关闭时，触发
     *
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        Long curTime = System.currentTimeMillis();
        Long userId = ThreadLocalCache.getUserId();
        log.info("用户下线，channel连接关闭！ 用户id:{}，channelId:{}", userId, ctx.channel().id().toString());
        if (!Objects.isNull(userId)) {
            // 从Map中 移除用户channel信息
            channelContextCommon.removeUserChannel(userId);
            // 移除redis中的心跳数据
            Key heartbeatKey = new ZhiChat(Keys.USER_HEARTBEAT);
            redisRepository.del(heartbeatKey, userId);
            // 跟新用户最后离线时间
            boolean b = userInfoDao.updateLastOffTimeByUserId(curTime, userId);
            if (!b) {
                throw CommonException.DB_MODIFY_COMMON_EXCEPTION;
            }
            // 删除 threadLocal
            ThreadLocalCache.clear();
        }
        super.channelInactive(ctx);
    }

    /**
     * 所有websocket请求过来，会先经过这里，所以可以在这里做【权限校验】
     * 1、先经过 channelActive()，再过这里，这俩个方法是channel建立连接(相当于登录时)触发。
     * 2、中间的所有消息走 channelRead0()。
     * 3、最后channel连接断开(相当于退出登录)走 channelInactive()
     *
     * @param ctx
     * @param evt
     * @throws Exception
     */
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        try {
            if (evt instanceof WebSocketServerProtocolHandler.HandshakeComplete) {
                log.info("握手事件开启");
                long curTime = System.currentTimeMillis();
                WebSocketServerProtocolHandler.HandshakeComplete complete = (WebSocketServerProtocolHandler.HandshakeComplete) evt;
//                log.info("request url:{}", complete.requestUri());
                // 获取请求头，获取token值
//                HttpHeaders entries = complete.requestHeaders();
//                String token = entries.get(Constants.AUTHORIZATION);
                String token = getToken(URLDecoder.decode(complete.requestUri(), "utf-8"));
                if (!StringUtils.hasText(token) || !zhiChatGlobalUtil.verifyToken(token) || Objects.isNull(zhiChatGlobalUtil.getPayload(token))) {
                    ctx.close();
                    log.info("token验证失败 - 无权访问");
                    return;
                }
                Long userId = zhiChatGlobalUtil.getPayload(token).getLong("uId");
                if (Objects.isNull(userId) || 0L == userId) {
                    ctx.close();
                    log.info("uId无效 - 无权访问");
                    return;
                }
                Key key = new ZhiChat(Keys.TOKEN);
                UserToken userToken = redisRepository.get(key, UserToken.class, userId);
                if (Objects.isNull(userToken) || !Objects.equals(userId, userToken.getUserId()) || !Objects.equals(token, userToken.getToken())) {
                    ctx.close();
                    log.info("uId验证失败 - 无权访问");
                    return;
                }
                // 放入 ThreadLocal cache
                ThreadLocalCache.putUserToken(userToken);
                // 连接成功，跟新心跳
                Key heartbeatKey = new ZhiChat(Keys.USER_HEARTBEAT, Duration.ofSeconds(Constants.READER_IDLE_TIME));
                redisRepository.set(heartbeatKey, curTime, userId);
                // 用户channel
                Channel userChannel = ctx.channel();
                // 放入 ChannelContext cache
                channelContextCommon.setUserChannel(userId, userChannel);
                // 获取用户自己所有群组，将用户channel放入群组 channel 中
                Key groupContactKey = new ZhiChat(Keys.CONTACT_GROUP);
                List<Long> groupContactIds = redisRepository.members(groupContactKey, Long.class, userId);
                if (!CollectionUtils.isEmpty(groupContactIds)) {
                    groupContactIds.forEach(groupId -> {
                        channelContextCommon.putAllUserChannelToGroupChannel(groupId, Lists.newArrayList(userChannel));
                    });
                }
                // 离线消息处理：
                // 1、查询用户表，获取用户最后离线时间 并做 处理（限制不能超过90天）
                UserInfoEntity sqlUserInfo = userInfoDao.selectById(userId);
                Long lastOffTime = sqlUserInfo.getLastOffTime();
                // 离线时间太久，超过90天，则只接收90天以内
                long compareTime = curTime - Constants.UNREAD_MESSAGE_QUERY_TIME_LIMIT;
                if (Objects.isNull(lastOffTime) || compareTime > lastOffTime) {
                    lastOffTime = compareTime;
                }
                // 2、查询用户所有会话信息
                List<UserSessionInfoVO> userSessionInfoList = chatUserSessionDao.getUserSessionInfoList(userId);
                userSessionInfoList.stream().filter(x -> Objects.equals(ContactTypeEnum.GROUP.getCode(), x.getReceiverType()))
                        .forEach(x -> {
                            LambdaQueryWrapper<ContactInfoEntity> getmemberCountQuery = Wrappers.lambdaQuery(ContactInfoEntity.class);
                            getmemberCountQuery.eq(ContactInfoEntity::getStatus, ContactStatusEnum.FRIEND.getCode())
                                    .eq(ContactInfoEntity::getContactId, x.getReceiverId());
                            int memberCount = Math.toIntExact(contactInfoDao.selectCount(getmemberCountQuery));
                            x.setMemberCount(memberCount);
                        });
                // 3、查询离线期间的消息（根据上次离线时间，距离当前时间大于90天，则获取90天以内的未读，反之取上去离线时间）
                List<UserChatMsgVO> userChatMsgVoList = getOfflineMsg(userId, lastOffTime);
                // 4、查询离线期间的好友申请数量（时间限制 同上）
                LambdaQueryWrapper<ApplyInfoEntity> applyQuery = Wrappers.lambdaQuery(ApplyInfoEntity.class);
                applyQuery.eq(ApplyInfoEntity::getReceiveId, userId)
                        .gt(ApplyInfoEntity::getLastApplyTime, lastOffTime)
                        .in(ApplyInfoEntity::getStatus, ApplyStatusEnum.APPLYING.getCode(), ApplyStatusEnum.EXPIRED.getCode());
                int applyCount = Math.toIntExact(applyInfoDao.selectCount(applyQuery));
                // 5、定义 用户连接ws初始化信息对象
                UserInitVO userInitVO = new UserInitVO();
                userInitVO.setUserSessionInfoList(userSessionInfoList);
                userInitVO.setChatMsgList(userChatMsgVoList);
                userInitVO.setApplyCount(applyCount);
                // 发送ws消息（给当前登录的用户发送初始化消息）
                MsgVO<UserInitVO> msgVO = new MsgVO<>();
                msgVO.setMsgType(MsgTypeEnum.ONLINE_INIT.getCode())
                        .setReceiverId(userId)
                        .setData(userInitVO);
                globalMsgHandle.sendMessageToUserChannel(msgVO, userId);
                log.info("握手完成");
            } else if (evt instanceof WebSocketServerProtocolHandler.ServerHandshakeStateEvent) {
                log.info("握手失败");
            } else if (evt instanceof IdleStateEvent) {
                log.info("连接超时");
            } else {
                log.info("未知事件:{}", evt);
            }
            super.userEventTriggered(ctx, evt);
        } catch (Exception e) {
            log.error("userEventTriggered error", e);
            ctx.close();
        }
    }

    // 获取离线期间的所有消息
    private List<UserChatMsgVO> getOfflineMsg(Long userId, Long searchTime) {
        // 1、在消息表，获取接收者是我 并且消息发送时间大于我上次离线的时间  的所有消息。。。消息状态是发送成功，避开 删除或者拉黑的人发来的消息（这种消息是发送不成功的状态的消息）
        // 2、在消息表，获取接收者是我的群组 并且消息发送时间大于我上次离线的时间 的所有消息。。。消息状态是发送成功，避开 被踢出群的人发送的不成功消息
        // （注意：只接收90天以内的未读消息，超过90天的不做提醒【90天是可设置的，例如：设置3年，就是读取三年内的未读消息】）
        List<Long> contactList = Lists.newArrayList(userId);
        Key groupContactKey = new ZhiChat(Keys.CONTACT_GROUP);
        List<Long> groupContactList = redisRepository.members(groupContactKey, Long.class, userId);
        if (!CollectionUtils.isEmpty(groupContactList)) {
            contactList.addAll(groupContactList);
        }
        Query mongoQuery = new Query(Criteria.where("receiverId").in(contactList)
                .and("sendTime").gt(searchTime)
                .and("status").is(ChatMsgStatusEnum.SUCCESS.getCode()));
//        mongoQuery.fields().include("senderId", "receiverId", "msg", "sendTime", "msgType");
        List<ChatMsgDoc> chatMsgDocs = mongoTemplate.find(mongoQuery, ChatMsgDoc.class);
        if (CollUtil.isEmpty(chatMsgDocs)) {
            return Lists.newArrayList();
        }
        return chatMsgDocs.stream().map(x -> {
            return new UserChatMsgVO()
                    .setMsgId(x.getId())
                    .setUserId(ThreadLocalCache.getUserId())
                    .setMsgType(x.getMsgType())
                    .setMsg(x.getMsg())
                    .setSessionId(x.getSessionId())
                    .setContactId(x.getReceiverId())
                    .setContactType(x.getReceiverType())
                    .setSendId(x.getSenderId().toString())
                    .setSendNickname(x.getSenderNickname())
                    .setSendTime(x.getSendTime().toString())
                    .setStatus(x.getStatus())
                    .setFileSize(Objects.isNull(x.getFileSize()) ? "0" : x.getFileSize().toString())
                    .setFileName(x.getFileName())
                    .setFileType(x.getFileType());
        }).collect(Collectors.toList());
    }

    private String getToken(String url) {
        if (StrUtil.isBlank(url) || !url.contains("?")) {
            return null;
        }
        String params = url.split("\\?")[1];
        if (StrUtil.isBlank(params) || !params.contains(Constants.AUTHORIZATION)) {
            return null;
        }
        String[] paramArray = params.split("&");
        if (paramArray.length == 0 || !paramArray[0].contains(Constants.AUTHORIZATION)
                || !StrUtil.equals(Constants.AUTHORIZATION, paramArray[0].split("=")[0])) {
            return null;
        }
        return paramArray[0].split("=")[1];
    }
}
